package cn.xval.ft.core.marketdata.manage;

import cn.xval.ft.core.ctp.CThostFtdcDepthMarketDataField;
import cn.xval.ft.grpc.MarketDataInfo;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

@Slf4j
public class DataWatcher implements Runnable {
    private final StreamObserver<MarketDataInfo> response;
    private final BlockingQueue<CThostFtdcDepthMarketDataField> queue = new ArrayBlockingQueue<>(240);
    private final Thread thread = new Thread(this);

    private int errorTimes = 0;

    public DataWatcher(StreamObserver<MarketDataInfo> response) {
        this.response = response;
        thread.start();
    }

    public int post(CThostFtdcDepthMarketDataField data) {
        try {
            queue.put(data);
            errorTimes = 0;
        }
        catch (InterruptedException ex) {
            errorTimes ++;
            log.error("派发数据错误", ex);
        }
        return errorTimes;
    }

    public void remove() {
        thread.interrupt();
    }

    @Override
    public void run() {
        try {
            while (!thread.isInterrupted()) {
                var data = queue.take();
                var info = MarketDataInfo.newBuilder()
                        .setTradingDay(data.getTradingDay())
                        .setExchangedId(data.getExchangeID())
                        .setLastPrice(data.getLastPrice())
                        .setPreSettlementPrice(data.getPreSettlementPrice())
                        .setPreClosePrice(data.getPreClosePrice())
                        .setPreOpenInterest(data.getPreOpenInterest())
                        .setOpenPrice(data.getOpenPrice())
                        .setHighestPrice(data.getHighestPrice())
                        .setLowerLimitPrice(data.getLowerLimitPrice())
                        .setVolume(data.getVolume())
                        .setTurnover(data.getTurnover())
                        .setOpenInterest(data.getOpenInterest())
                        .setClosePrice(data.getClosePrice())
                        .setSettlementPrice(data.getSettlementPrice())
                        .setUpperLimitPrice(data.getUpperLimitPrice())
                        .setLowerLimitPrice(data.getLowerLimitPrice())
                        .setPreDelta(data.getPreDelta())
                        .setCurrDelta(data.getCurrDelta())
                        .setUpdateTime(data.getUpdateTime())
                        .setUpdateMillisec(data.getUpdateMillisec())
                        .setBidPrice1(data.getBidPrice1())
                        .setBidVolume1(data.getBidVolume1())
                        .setAskPrice1(data.getAskPrice1())
                        .setAskVolume1(data.getAskVolume1())
                        .setBidPrice2(data.getBidPrice2())
                        .setBidVolume2(data.getBidVolume2())
                        .setAskPrice2(data.getAskPrice2())
                        .setAskVolume2(data.getAskVolume2())
                        .setBidPrice3(data.getBidPrice3())
                        .setBidVolume3(data.getBidVolume3())
                        .setAskPrice3(data.getAskPrice3())
                        .setAskVolume3(data.getAskVolume3())
                        .setBidPrice4(data.getBidPrice4())
                        .setBidVolume4(data.getBidVolume4())
                        .setAskPrice4(data.getAskPrice4())
                        .setAskVolume4(data.getAskVolume4())
                        .setBidPrice5(data.getBidPrice5())
                        .setBidVolume5(data.getBidVolume5())
                        .setAskPrice5(data.getAskPrice5())
                        .setAskVolume5(data.getAskVolume5())
                        .setAveragePrice(data.getAveragePrice())
                        .setActionDay(data.getActionDay())
                        .setInstrumentId(data.getInstrumentID())
                        .setExchangeInstId(data.getExchangeInstID())
                        .setBandingUpperPrice(data.getBandingUpperPrice())
                        .setBandingLowerPrice(data.getBandingLowerPrice())
                        .build();
                response.onNext(info);
            }
        } catch (InterruptedException e) {
            log.info("停止接受数据");
        }
        SpiManager.removeDataWatcher(this);
        response.onCompleted();
    }
}
